package com.glodon.paas.job.agent.mq;

import static com.glodon.paas.job.JobConstants.*;
import static com.glodon.paas.job.model.TaskStatus.*;

import java.io.File;
import java.net.URL;
import java.net.URLClassLoader;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;

import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.commons.collections.map.LRUMap;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;

import com.glodon.paas.job.agent.Agent;
import com.glodon.paas.job.agent.task.AbstractTaskListener;
import com.glodon.paas.job.agent.task.TaskListener;
import com.glodon.paas.job.agent.worker.DefaultWorkerFactory;
import com.glodon.paas.job.agent.worker.WorkerConfig;
import com.glodon.paas.job.agent.worker.WorkerFactory;
import com.glodon.paas.job.model.Task;
import com.glodon.paas.job.util.JobUtil;

public class JmsAgent implements Agent {
    private static final Logger logger = LoggerFactory.getLogger(JmsAgent.class);
    // five minutes
    private static final int LAST_MONITOR_SECOND = 5 * 60;
    private static final int MAX_FAIL_COUNT = 5;
    private int maxFailCount = MAX_FAIL_COUNT;
    // save max task status
    private static final int MAX_STATUS = 1000;
    private JmsTemplate jmsTemplate;
    private AgentHeartBeat heartBeat;
    private ConnectionFactory connectionFactory;
    private ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
    private Map<String, LRUMap> taskStatus = new HashMap<String, LRUMap>();
    private WorkerConfig[] workerConfigs;
    private MultipleDestinationMessageListenerContainer container;
    private WorkerFactory workerFactory = new DefaultWorkerFactory();
    private int workerConcurrent = 1;
    private Properties conf = new Properties();
    private Task task;
    public JmsAgent() {
        loadProperties("job-agent.properties");
        loadProperties("job-agent-global.properties");
        String url = conf.getProperty("activemq.url", "tcp://localhost:61616");
        String username = conf.getProperty("activemq.username", "");
        String password = conf.getProperty("activemq.password", "");
        String maxFailCountStr = conf.getProperty("max.fail.count");
        String workerTypesStr = conf.getProperty("worker.types", "");
        String workerCmdsStr = conf.getProperty("worker.cmds", "");
        String workerEnvpStr = conf.getProperty("worker.envp", "");
        logger.info("worker types: {}", workerTypesStr);
        logger.info("worker cmds: {}", workerCmdsStr);
        logger.info("worker envp: {}", workerEnvpStr);
        if (StringUtils.isNotBlank(workerTypesStr)) {
            String[] workerTypes = StringUtils.splitPreserveAllTokens(workerTypesStr, '|');
            String[] workerCmds = StringUtils.splitPreserveAllTokens(workerCmdsStr, '|');
            String[] workerEnvps = StringUtils.splitPreserveAllTokens(workerEnvpStr, '|');
            workerConfigs = new WorkerConfig[workerTypes.length];
            for (int i = 0; i < workerTypes.length; i++) {
                String workerType = workerTypes[i];
                String workerCmd = workerCmds[i];
                String[] envp = StringUtils.split(workerEnvps[i], ':');
                workerConfigs[i] = new WorkerConfig(workerType, workerCmd, envp);
            }
        }
        
        if (maxFailCountStr != null)
            maxFailCount = Integer.parseInt(maxFailCountStr);
        connectionFactory = new ActiveMQConnectionFactory(username, password, url);
    }

    private void loadProperties(String path) {
        try {
            if (loadConfigFromJarFolder(path))
                return;
            conf.load(Thread.currentThread().getContextClassLoader().getResourceAsStream(path));
            logger.info("classpath config file {} success", path);
        } catch (Exception e) {
            logger.info("classpath config file {} not found", path);
            // ignores
        }
    }

    private boolean loadConfigFromJarFolder(String path) {
        try {
            File jar = new File(JmsAgent.class.getProtectionDomain().getCodeSource().getLocation().getFile());
            URLClassLoader currentFolderCL = new URLClassLoader(new URL[] { jar.getParentFile().toURI().toURL() });
            conf.load(currentFolderCL.getResourceAsStream(path));
            logger.info("current folder config file {} success", path);
            return true;
        } catch (Exception e) {
            logger.info("current folder config file {} not found", path);
            // ignores
            return false;
        }
    }

    @Override
    public void start() {
        jmsTemplate = new JmsTemplate(connectionFactory);
        jmsTemplate.setDefaultDestinationName(AGENT_RESULT_QUEUE);
        listenMultiTaskQueue();
        heartBeat = new AgentHeartBeat(this);
        heartBeat.start();
    }

    private void listenMultiTaskQueue() {
        container = new MultipleDestinationMessageListenerContainer();
        container.setSessionAcknowledgeMode(Session.CLIENT_ACKNOWLEDGE);
        container.setConnectionFactory(connectionFactory);
        container.setDestination(new MultipleDestination(workerConfigs));
        container.setMessageListener(createTaskListener(workerFactory));
        container.setMaxConcurrentConsumers(workerConcurrent);
        container.initialize();
        container.start();
    }

    private AbstractTaskListener createTaskListener(WorkerFactory workerFactory) {
        AbstractTaskListener taskListener = new TaskListener();
        taskListener.setWorkerFactory(workerFactory);
        taskListener.setWorkerConfigs(workerConfigs);
        taskListener.setAgent(this);
        return taskListener;
    }

    @Override
    public void sendMessage(final Task task) {
        logger.info("send result message, task: {}", task);
        addTaskStatus(task);
        // get task results in last five minutes
        checkLastFailure(task.getTaskType());
        jmsTemplate.send(new MessageCreator() {
            @Override
            public Message createMessage(Session session) throws JMSException {
                return task.toMessage(session);
            }
        });
        this.setTask(task);
        System.out.println(task.toJson());
        JobUtil.sendJobStatus(task);
    }

    private void checkLastFailure(String type) {
        ReadLock readLock = lock.readLock();
        try {
            readLock.lock();
            int failCount = getFailTaskCount(type, LAST_MONITOR_SECOND);
            if (maxFailCount > 0 && failCount >= maxFailCount) {
                // stop the listener
                logger.error("in last {} second, failure count great then {}, then stop the [type: {}] listener",
                        new Object[] { LAST_MONITOR_SECOND, MAX_FAIL_COUNT, type });
                container.remove(type);
            }
        } finally {
            readLock.unlock();
        }
    }

    private int getFailTaskCount(String type, int second) {
        int count = 0;
        LRUMap status = getStatusMap(type);
        Set<Entry<Date, Task>> entrySet = status.entrySet();
        for (Entry<Date, Task> entry : entrySet) {
            Date endTime = entry.getKey();
            Task task = entry.getValue();
            Calendar cal = Calendar.getInstance();
            cal.add(Calendar.SECOND, -second);
            if (endTime.after(cal.getTime()) && task.getStatus() == failure) {
                count++;
            }
        }
        return count;
    }

    private void addTaskStatus(Task task) {
        if (task.getStatus() != success && task.getStatus() != failure)
            return;
        String type = task.getTaskType();
        WriteLock writeLock = lock.writeLock();
        try {
            writeLock.lock();
            LRUMap status = getStatusMap(type);
            status.put(new Date(), task);
        } finally {
            writeLock.unlock();
        }
    }

    private LRUMap getStatusMap(String type) {
        LRUMap status = taskStatus.get(type);
        if (status == null) {
            status = new LRUMap(MAX_STATUS);
            taskStatus.put(type, status);
        }
        return status;
    }

    @Override
    public void stop() {
        heartBeat.stop();
        container.shutdown();
        container.stop();
    }

    @Override
    public void setWorkerConfigs(WorkerConfig[] workerConfigs) {
        this.workerConfigs = workerConfigs;
    }

    public WorkerConfig[] getWorkerConfigs() {
        return workerConfigs;
    }

    public void setWorkerFactory(WorkerFactory workerFactory) {
        this.workerFactory = workerFactory;
    }

	public Task getTask() {
		return task;
	}

	public void setTask(Task task) {
		this.task = task;
	}


}
